package marsdonne.tasks;

import org.apache.flink.api.common.io.OutputFormat;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;

public class SqlTask implements Serializable {

    public enum TestResult {
        Passed, Failed
    }

    public enum DbType {
        PgSql("org.postgresql.Driver|jdbc:postgresql://192.168.0.198:5432/stat-examples?user=postgres&password=postgres"),
        H2("org.h2.Driver|jdbc:h2:mem:stat-examples;MVCC=TRUE;LOCK_TIMEOUT=10000;DB_CLOSE_ON_EXIT=FALSE");

        private final String driver;

        DbType(String driver) {
            this.driver = driver;
        }
    }

    private final JDBCOutputFormat jdbcOutput;
    private final int pCounter;

    private final boolean testable;

    private final List<String> periods;

    public SqlTask(DbType dbType, String sql, int[] typesArray, String periods) {
        this(false, dbType, sql, typesArray, periods);
    }

    public SqlTask(boolean testable, DbType dbType, String sql, int[] typesArray, String periods) {
        this.periods = new LinkedList<>(Arrays.asList(periods.split("\\|")));
        this.testable = testable;
        pCounter = typesArray.length;

        String[] jdbcConfig = dbType.driver.split("\\|");

        jdbcOutput = JDBCOutputFormat.buildJDBCOutputFormat()
                .setDrivername(jdbcConfig[0])
                .setDBUrl(jdbcConfig[1])
                .setQuery(sql)
                .setSqlTypes(typesArray)
                .finish();
    }

    private static class TestOutput implements OutputFormat<Row> {

        // must be static
        public static final List<Row> values = new LinkedList<>();

        @Override
        public void configure(Configuration configuration) {

        }

        @Override
        public void open(int i, int i1) throws IOException {

        }

        @Override
        public void writeRecord(Row o) throws IOException {
            values.add(o);
        }

        @Override
        public void close() throws IOException {

        }
    }

    public TestResult submitTo(ExecutionEnvironment env) throws Exception {
        DataSet ds = env.fromCollection(periods)
                .flatMap((String triggered, Collector<Row> collector) -> {
                    String[] range = triggered.split(",");
                    System.out.println("triggered:" + range[0] + " >>> " + range[1]);
                    Row row = new Row(pCounter);
                    if (pCounter > 0) {
                        row.setField(0, range[0]);
                        row.setField(1, range[0]);
                        row.setField(2, range[1]);
//                        collector.collect(row);//合规时，转发结果数据到后续环节
                    }
                    collector.collect(row);//仅为本例兼容测试代码
                })
                .returns(Row.class)
                .output(jdbcOutput)
                .getDataSet();

        if (this.testable) {
            TestOutput.values.clear();
            ds.output(new TestOutput()).getDataSet().count();
            return TestOutput.values.size() == periods.size() ? TestResult.Passed : TestResult.Failed;
        } else {
            ds.count();
            return null;
        }

    }
}
